package service

import (
	"fmt"
	"gitee.com/sunny-zhangqiang/zcache_new/cache/base"
	"gitee.com/sunny-zhangqiang/zcache_new/comm"
	"github.com/HuKeping/rbtree"
	"strconv"
	"sync"
	"sync/atomic"
	"time"
)

const (
	slot = 64
)

var (
	mge             *merge
	mergeExpireData = mergeData{
		false, false, nil, "", mergeNodeExpireErr,
	}
	mergeClientAbortData = mergeData{
		false, false, nil, "", mergeNodeClientAbortErr,
	}
)

type mergeData struct {
	clientNotify     bool
	cachable         bool
	value            []byte
	revalidateStatus string
	err              error
}

type mergeNode struct {
	key         string
	clientToken string
	expireTime  int64
	memAddress  string
	mergeToken  string
	readers     []chan mergeData
}

func (node *mergeNode) Less(than rbtree.Item) bool {
	thanNode := than.(*mergeNode)

	if node.expireTime < thanNode.expireTime {
		return true
	}

	if node.expireTime > thanNode.expireTime {
		return false
	}

	return node.memAddress < thanNode.memAddress
}

type oneClientMerge struct {
	sync.Mutex
	// map[clientToken] --> map[sourceKey]*mergeNode
	// 对应的clientAbort时, 将向该clientToken上的所有mergeNode发送Abort通知
	m map[string]map[string]*mergeNode
}

func (ocm *oneClientMerge) insert(node *mergeNode) {
	ocm.Lock()
	defer ocm.Unlock()

	ctm := ocm.m[node.clientToken]
	if ctm == nil {
		ctm = make(map[string]*mergeNode)
		ocm.m[node.clientToken] = ctm
	}

	ctm[node.key] = node
}

func (ocm *oneClientMerge) delete(node *mergeNode) {
	ocm.Lock()
	defer ocm.Unlock()

	ctm := ocm.m[node.clientToken]
	if ctm == nil {
		return
	}

	delete(ctm, node.key)
}

type oneMerge struct {
	sync.Mutex
	m       map[string]*mergeNode // map[sourceKey] = mergeNode
	expireM *rbtree.Rbtree        // map[expireTime] = mergeNode
}

type merge struct {
	version int64
	oms     [slot]oneMerge
	ocms    [slot]oneClientMerge
}

func (mge *merge) getOneMerge(key *base.ObjectKey) *oneMerge {
	return &mge.oms[key.Hash%slot]
}

func (mge *merge) getOneClientMerge(clientToken string) *oneClientMerge {
	hash := comm.FnvHash(comm.String2Bytes(clientToken))
	return &mge.ocms[hash%slot]
}

func (mge *merge) wait(key *base.ObjectKey, clientToken string,
	mergeTimeoutMSec int64) (ch chan mergeData, mergeToken string) {

	ch = nil
	mergeToken = ""

	om := mge.getOneMerge(key)
	ocm := mge.getOneClientMerge(clientToken)

	strKey := comm.Bytes2String(key.Key)

	om.Lock()
	node := om.m[strKey]
	if node == nil {
		node = &mergeNode{
			key:         strKey,
			clientToken: clientToken,
			expireTime:  time.Now().UnixMilli() + mergeTimeoutMSec,
			mergeToken: serverToken + "-" +
				strconv.FormatInt(atomic.AddInt64(&mge.version, 1), 16),
			readers: make([]chan mergeData, 0, 4),
		}
		node.memAddress = fmt.Sprintf("%p", node)
		om.m[strKey] = node
		om.expireM.Insert(node)
		ocm.insert(node)
		mergeToken = node.mergeToken
	} else {
		ch = make(chan mergeData, 1)
		node.readers = append(node.readers, ch)
	}
	om.Unlock()

	return ch, mergeToken
}

func (mge *merge) notify(key *base.ObjectKey, md mergeData, mergeToken string) {
	om := mge.getOneMerge(key)
	strKey := string(key.Key)

	om.Lock()
	node := om.m[strKey]
	if node != nil && node.mergeToken == mergeToken {
		for i := 0; i < len(node.readers); i++ {
			node.readers[i] <- md
			node.readers[i] = nil
		}
		node.readers = node.readers[:0]
		om.expireM.Delete(node)
		delete(om.m, strKey)
		ocm := mge.getOneClientMerge(node.clientToken)
		ocm.delete(node)
	}
	om.Unlock()
}

func (mge *merge) clientAbort(clientToken string) {
	ocm := mge.getOneClientMerge(clientToken)

	tmp := make(map[string]*mergeNode)

	ocm.Lock()
	ctm := ocm.m[clientToken]
	for k, v := range ctm {
		tmp[k] = v
	}
	ocm.Unlock()

	for k, v := range tmp {
		key := base.GenObjectKey(comm.String2Bytes(k))
		mge.notify(key, mergeClientAbortData, v.mergeToken)
	}
}

func (mge *merge) checkExpire(i int) {
	om := &mge.oms[i]
	om.Lock()
	defer om.Unlock()

	for {
		it := om.expireM.Min()
		if it == nil {
			return
		}

		node := it.(*mergeNode)
		if node.expireTime > time.Now().UnixMilli() {
			return
		}

		om.expireM.Delete(node)
		mNode := om.m[node.key]
		if mNode == nil || mNode.memAddress != node.memAddress {
			continue
		}

		for i := 0; i < len(node.readers); i++ {
			node.readers[i] <- mergeExpireData
			node.readers[i] = nil
		}
		node.readers = node.readers[:0]

		delete(om.m, node.key)
		ocm := mge.getOneClientMerge(node.clientToken)
		ocm.delete(node)
	}
}

func (mge *merge) expireLoop() {
	ticker := time.NewTicker(time.Second)
	for {
		<-ticker.C
		for i := 0; i < slot; i++ {
			mge.checkExpire(i)
		}
	}
}

func openMerge() {
	mge = &merge{}
	for i := 0; i < slot; i++ {
		mge.oms[i].m = make(map[string]*mergeNode)
		mge.oms[i].expireM = rbtree.New()
		mge.ocms[i].m = make(map[string]map[string]*mergeNode)
	}
	go mge.expireLoop()
}
